package newStreaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object DriverHA extends App {
  val sparkConf = new SparkConf()
  sparkConf.setAppName("wordCountDS")
  //  sparkConf.setMaster("local[4]")
  val sc = new SparkContext(sparkConf)
  val checkpointPath = "data/DriverHA"

  def createContext(): StreamingContext = {
    println("*" * 100)
    println("create StreamingContext")
    println("*" * 100)
    val ssc = new StreamingContext(sc, Durations.seconds(5))
    ssc.checkpoint(checkpointPath)
    val ds = ssc.socketTextStream("node2", 9999)
    ds.map((_, 1)).reduceByKey(_ + _).print()

    ssc
  }

  //  ssc.checkpoint("data/windowSlide")
  //  // 先去node1 启动 nc -lk 9999
  //  val ds = ssc.socketTextStream("node2", 9999)
  ////  val ssc = new StreamingContext(sparkContext, Durations.seconds(10))
  val ssc = StreamingContext.getOrCreate(checkpointPath, createContext)

  ssc.start()
  ssc.awaitTermination()
  ssc.stop()

  /**
   * spark-submit --master yarn-cluster --supervise --executor-memory 512m  --total-executor-cores 2  SparkDemo-1.0-SNAPSHOT.jar
   */

}
